package middleware.mq;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.TimeUnit;

/**
 * 发生异步信息
 *
 * 场景：
 * 在对响应时间 敏感的业务场景
 * 发送端不能容忍长时间的等待
 *
 * Author: shen qi han
 * Date: 2022/4/3 下午10:25
 */
public class AsyncProducerMain {

    public static void main(String[] args) throws Exception {
        // init producer
        DefaultMQProducer producer = new DefaultMQProducer("TopicProducer");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);

        // logic
        int msgCnt = 100;
        final CountDownLatch2 count = new CountDownLatch2(msgCnt);
        for(int i = 0; i < msgCnt; i++){
            final int index = i;
            // msg
            Message msg = new Message("TopicDL",
                    "DL Method",
                    "DL Data".getBytes(RemotingHelper.DEFAULT_CHARSET));
            // recv async result call
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    count.countDown();
                    System.out.printf("%-10d OK %s %n",
                            index, sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    count.countDown();
                    System.out.printf("%-10d Exception %s %n", index, e);
                }
            });

        }

        // wait 5s
        count.await(5, TimeUnit.SECONDS);
        // shut down
        producer.shutdown();
    }

}
